#include <Active.h>



Active::Active(uint l_ParallelThreads): 
done( new Message ) ,
m_ParallelThreads(l_ParallelThreads),
thdGroup(new boost::thread_group())
{
  
  for (uint i=0; i< m_ParallelThreads;i++)
    thdGroup->add_thread (new boost::thread( boost::bind(&Active::Run, this) ) );
  
}

Active::~Active() 
{
  for (uint i=0; i<m_ParallelThreads;i++)
    Send( done );
  
  //std::cout << "Joining Threads" << std::endl;
  thdGroup->join_all();
  //std::cout << "Threads Joined" << std::endl;
  
}
  
void Active::Run() 
{
  MessagePtr msg;  
  do
  {
    mq.wait_and_pop(msg);
    msg->Execute();      
  }while(msg != done);  
}

 // Enqueue a message
void Active::Send( MessagePtr m ) 
{
  mq.push( m );
} 



Active::Message::~Message()
{}
void Active::Message::Execute()
{}




